package com.xflib.framework.ratelimit.core;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class LeakyBucketExample {
    private final Queue<Integer> bucket = new LinkedList<>(); // 漏桶队列
    private final int capacity; // 漏桶容量
    private final int leakRate; // 漏出速率（单位：请求/秒）
    private final Object lock = new Object(); // 锁对象

    public LeakyBucketExample(int capacity, int leakRate) {
        this.capacity = capacity;
        this.leakRate = leakRate;
        startLeaking(); // 启动漏出线程
    }

    // 请求入队（生产者）
    public boolean addRequest(int request) {
        synchronized (lock) {
            if (bucket.size() >= capacity) {
                System.out.println("漏桶已满，抛弃请求: " + request);
                return false; // 漏桶已满，抛弃请求
            }
            bucket.offer(request);
            System.out.println("接收请求: " + request + "，当前桶大小: " + bucket.size());
            lock.notifyAll(); // 唤醒漏出线程
            return true;
        }
    }

    // 启动漏出线程（消费者）
    private void startLeaking() {
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        scheduler.scheduleAtFixedRate(() -> {
            synchronized (lock) {
                if (!bucket.isEmpty()) {
                    int request = bucket.poll();
                    System.out.println("处理请求: " + request + "，剩余桶大小: " + bucket.size());
                }
            }
        }, 0, 1000 / leakRate, TimeUnit.MILLISECONDS); // 固定速率漏出
    }

    // 测试程序
    public static void main(String[] args) {
        LeakyBucketExample leakyBucket = new LeakyBucketExample(10, 2); // 容量10，漏出速率2请求/秒

        // 模拟生产者线程
        Thread producerThread = new Thread(() -> {
            int request = 0;
            while (!Thread.currentThread().isInterrupted()) {
                leakyBucket.addRequest(request++);
                try {
                    Thread.sleep(300); // 模拟请求间隔
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });

        producerThread.start();

        // 运行10秒后停止
        try {
            Thread.sleep(10000);
            producerThread.interrupt();
            producerThread.join();
            System.out.println("测试结束");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}